/*
 * Copyright (c) 2017, 1DAOYUN and/or its affiliates. All rights reserved.
 * 1DAOYUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package com.iot.nev.gateway.router.kafka;

import java.math.BigDecimal;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;

/**
 * 广播信息被处理后的的反馈（只要真的异步操作）。
 * 
 * 
 * @author Song Xueyong
 */
public class KafkaProducerCallback implements Callback {

	private Logger logger = Logger.getLogger(KafkaProducerCallback.class);
	private final long startTime;
	private final long messageNo;
	private final String message;

	/**
	 * Callback
	 * 
	 * @param startTime
	 * @param messageNo
	 * @param message
	 */
	public KafkaProducerCallback(long startTime, long messageNo, String message) {
		this.startTime = startTime;
		this.messageNo = messageNo;
		this.message = message;
	}

	/**
	 * 反馈处理。
	 *
	 * @param metadata
	 *            The metadata for the record that was sent (i.e. the partition and
	 *            offset). Null if an error occurred.
	 * @param exception
	 *            The exception thrown during processing of this record. Null if no
	 *            error occurred.
	 */
	@Override
	public void onCompletion(RecordMetadata metadata, Exception exception) {
		double duration = new BigDecimal(System.currentTimeMillis() - startTime).divide(new BigDecimal(1000)).doubleValue();
		if (metadata != null) {
			logger.info("## Message:" + messageNo + " callback duration:" + duration + "s");
		} else {
			logger.info("## Message Error:" + messageNo + "Exception:" + exception + " callback duration:" + duration + "s");
		}

	}

}
